package com.hz.cheetah.hub.zookeeper;

import com.hz.cheetah.hub.ServiceRegistry;
import com.hz.cheetah.hub.StateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于 ZooKeeper 的服务注册接口实现
 *
 * @author yangchuan
 * @since 1.0.0
 */
public class ZooKeeperServiceRegistry implements ServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceRegistry.class);

    private  ZookeeperClient zkClient;

    private String zkAddress;

    private final ConcurrentHashMap<String,String> registered = new ConcurrentHashMap<String,String>();

    public ZooKeeperServiceRegistry() {
    }

    public ZooKeeperServiceRegistry(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    public void setZkAddress(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    public void init(){
        // 创建 ZooKeeper 客户端
        zkClient = new ZookeeperClientImp(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        LOGGER.debug("connect zookeeper");
        zkClient.addStateListener(new StateListener() {
            @Override
            public void stateChanged(int connected) {
                if (connected == StateListener.RECONNECTED) {
                    LOGGER.debug(".................reconnected zookeeper............");
                    recover();
                }
            }
        });
    }

    @Override
    public void register(String serviceName, String serviceAddress) {
        registerToZK(serviceName, serviceAddress);
        registered.put(serviceName, serviceAddress);
    }

    /**
     * 重新加载节点
     */
    protected void recover(){
        Iterator<Map.Entry<String,String>> entries = registered.entrySet().iterator();
        Map.Entry<String,String> entry = null;
        while(entries.hasNext()){
            entry = entries.next();
            registerToZK(entry.getKey(),entry.getValue());
        }
    }

    protected void registerToZK(String serviceName, String serviceAddress) {
        // 创建 registry 节点（持久）
        String path =Constant.ZK_ROOT_PATH;
        String servers = Constant.SERVICE_SERVERS;

        createPersistent(path);
        createPersistent(servers);
        servers = servers+"/"+getURLCodeString(serviceAddress);
        if(!zkClient.exists(servers))
            zkClient.createEphemeral(servers, null);
        servers = null;
        path = path + "/" + serviceName;
        createPersistent(path);
        path = path + Constant.SERVICE_PROVIDERS ;
        createPersistent(path);

        // 创建 address 节点（临时）
        String temppath = getURLCodeString(Constant.PROTOCOL + "://" + serviceAddress + "/?weight=100");
        path = path + "/"+temppath;
        zkClient.createEphemeral(path, null);
        LOGGER.debug("create address node: {}", temppath);
        path = null;
    }

    private String getURLCodeString(String string) {
        try {
            return  URLEncoder.encode(string, "utf-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void createPersistent(String path){
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path);
            LOGGER.debug("create path node: {}", path);
        }
    }
}